package app.redis;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import app.htby.utils.KafkaConsumeUtils;
import app.htby.utils.RedisUtils;
import app.htby.utils.SpringContextUtil;
import app.kafka.consume.DisWebConsumer;
import app.kafka.consume.DisWebConsumerRunnable;
import redis.clients.jedis.Jedis;

/**
 * 大数据redis的重连
 * 
 * @author lfy.xys
 * @date 2018年6月5日
 *
 */
public class ReconnDataPlusRedis {

	private static Logger logger = LoggerFactory.getLogger(ReconnDataPlusRedis.class);
	public static Lock reconnDataPlusRedisLock = new ReentrantLock();
	
	private static ReconnDataPlusRedis reconnDataPlusRedis= null;
	
	private ReconnDataPlusRedis() {}
	
	// 是否连接成功 
	public static boolean IS_flag = true;
	
	// 当前连接成功 标志
	public static boolean flag = false;
	
	static {
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				while(true) {
					try {
						Thread.sleep(10 * 1000);
						//如果 连接成功，则跳过。否则重新连接
						if(!IS_flag) {
							ReconnDataPlusRedis.getInstance().reconn();
						}
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				
			}
		}).start();
	}
	
	/**
	 * 单例模式
	 */
	public static synchronized ReconnDataPlusRedis getInstance() {
		if (reconnDataPlusRedis == null) {
			reconnDataPlusRedis = new ReconnDataPlusRedis();
		}
		
		return reconnDataPlusRedis;
	}

	/**
	 * 重新连接
	 * @author lfy.xys
	 * @date 2018年6月5日
	 *
	 */
	public void reconn() {

		// 使用锁，放置创建多个，后面遇到有锁 的等待
		while(reconnDataPlusRedisLock.tryLock()) {
			//得到了锁
			try {
				while (!flag) {
					System.out.println("-------------------flag: " + flag  );
					try {
						// 等待时间
						logger.info("正在重连，等待时间...{}", RedisUtils.reconnectTime);
						Thread.sleep(RedisUtils.reconnectTime);
						Jedis jedis = JedisDataPlusPool.getJedis();
						// 判断是否连接成功
						flag = true;
						JedisDataPlusPool.returnRes(jedis);
						KafkaConsumeUtils.IS_CONSUMER_DIS = true;
						IS_flag = true;
						
						logger.info("分发kafka的topic设置继续消费.");

						if (KafkaConsumeUtils.IS_CONSUMER_DIS) {
							logger.info("topic的set个数：{} 个", KafkaConsumeUtils.topicSet.size());
							// 重新消费
							for (String topic : KafkaConsumeUtils.topicSet) {
								// 没有被消费过的topic
								String beanId = "Consumer_" + KafkaConsumeUtils.topicCount++;// 设置id
								SpringContextUtil.registerBean(beanId, DisWebConsumer.class.getName());// 使用spring创建bean
								// 创建线程消费
								new Thread(new DisWebConsumerRunnable(topic, beanId)).start();
							}
						}

						logger.info("连接上大数据的redis.");
					} catch (Exception e) {
						logger.error("没有连接上大数据的redis，正在重新连接...");
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				reconnDataPlusRedisLock.unlock();
			}
			
		} 
		
	}
}
